package cn.xinfei.xdecision.pulsar.consumer;

import cn.xinfei.xdecision.conf.ApplicationContextUtil;
import cn.xinfei.xdecision.listener.XMultApplicationEventListener;
import cn.xinfei.xdecision.listener.event.CustomerStopEvent;
import cn.xinfei.xdecision.pulsar.util.PulsarUtils;
import cn.xinfei.xdecision.thread.CustomerThreadPoolExecutor;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.ConsumerStats;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationEvent;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import java.lang.reflect.Method;
import java.util.*;
import java.util.concurrent.CopyOnWriteArrayList;

@Slf4j
@ConditionalOnProperty(name = "spring.xpulsar.enable")
@Component("XPulsarConsumerRegister")
//@ConditionalOnBean(XPulsarClientFactory.class)
@Order(Integer.MAX_VALUE)
public class XPulsarConsumerRegister extends XMultApplicationEventListener {

    /**
     * mq的消费者线程池不需要统一配置管理
     */
    @Resource(name = "pulsarConsumerThreadPool")
    private CustomerThreadPoolExecutor pulsarConsumerThreadPool;


    public static CopyOnWriteArrayList<XPulsarConsumerThread> XPulsarConsumerThreadContainer = new CopyOnWriteArrayList<XPulsarConsumerThread>();


    @Override
    protected Class<?>[] listenerEvent() {
        return new Class[]{ApplicationReadyEvent.class, CustomerStopEvent.class};
    }

    @Override
    protected void onXApplicationEvent(ApplicationEvent applicationEvent) {
        if (applicationEvent instanceof ApplicationReadyEvent) {
            log.warn("==========XPulsarConsumerRegisterEvent================{}", applicationEvent);
            startPulsarConsumer(null);
        } else if (applicationEvent instanceof CustomerStopEvent) {
            log.warn("==========XPulsarConsumerStopEvent================{}", applicationEvent);
            stop();
        } else {
            log.warn("==========applicationEvent================{}", applicationEvent);
        }
    }

    @SuppressWarnings({"unchecked", "rawtypes"})
    public static List<Method> getClassMethodByAnnotation(Class clz, Class annoClz) throws Exception {

        clz = Class.forName(clz.getName(), true, clz.getClassLoader());
        List<Method> result = new ArrayList<Method>();
        for (Method method : clz.getMethods()) {
            if (method.getAnnotation(annoClz) != null) {
                result.add(clz.getMethod(method.getName(), method.getParameterTypes()));
            }
        }
        return result;
    }


    public void startPulsarConsumer(String topic) {

        log.info("==========CustomerStartedEvent===XPulsarConsumerRegister================");
        try {
            ApplicationContext applicationContext = ApplicationContextUtil.getApplicationContext();
            Map<Object, List<Method>> methodMap = new HashMap<>();
            if (applicationContext != null) {
                Map<String, Object> xPulsarConsumer = applicationContext.getBeansWithAnnotation(Service.class);
                if (CollectionUtils.isEmpty(xPulsarConsumer) == false) {
                    Set<Map.Entry<String, Object>> entries = xPulsarConsumer.entrySet();
                    Iterator<Map.Entry<String, Object>> iterator = entries.iterator();
                    while (iterator.hasNext()) {
                        Map.Entry<String, Object> next = iterator.next();
                        String key = next.getKey();
                        Object value = next.getValue();
                        List<Method> methodList = getClassMethodByAnnotation(value.getClass(), XPulsarConsumerAnnotate.class);
                        if (null != methodList && methodList.size() > 0) {
                            methodMap.put(value, methodList);
                        }
                    }
                }
            }
            if (CollectionUtils.isEmpty(methodMap) == false) {
                Set<Map.Entry<Object, List<Method>>> entries = methodMap.entrySet();
                Iterator<Map.Entry<Object, List<Method>>> iterator = entries.iterator();
                while (iterator.hasNext()) {
                    Map.Entry<Object, List<Method>> next = iterator.next();
                    List<Method> methodList = next.getValue();
                    Object obj = next.getKey();
                    for (int i = 0; i < methodList.size(); i++) {
                        Method method = methodList.get(i);
                        int concurrent = getConcurrent(method);
                        for (int j = 0; j < concurrent; j++) {
                            XPulsarConsumerThread consumerThread = new XPulsarConsumerThread(method, obj, j);
                            if (!StringUtils.isEmpty(topic) && topic.equals(consumerThread.getTopic()) == false) {
                                log.info("continue start consumer. topic={}", topic);
                                continue;
                            }
                            log.info("startPulsarListener register start. topic={}", topic);
                            Thread thread = new Thread(consumerThread);
                            thread.setName(consumerThread.getConsumerName() + "_" + consumerThread.getTopic() + "_" + consumerThread.getConcurrentNum());
                            thread.start();
                            XPulsarConsumerThreadContainer.add(consumerThread);
                            log.info("startPulsarListener register end. topic={}", topic);
                        }
                    }
                }
            }
        } catch (Throwable throwable) {
            throwable.printStackTrace();
        }
    }

    private int getConcurrent(Method method) {

        try {
            XPulsarConsumerAnnotate annotation = method.getAnnotation(XPulsarConsumerAnnotate.class);
            String concurrent = annotation.concurrent();
            if (concurrent.startsWith("$")) {
                return Integer.parseInt(PulsarUtils.getValueFromEnv(concurrent));
            } else {
                return Integer.parseInt(concurrent);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return 1;
    }

    public void stop(String topic) {
        log.info("stop pulsar consumer start . topic={}", topic);
        for (int i = XPulsarConsumerThreadContainer.size() - 1; i >= 0; i--) {
            XPulsarConsumerThread xPulsarConsumerThread = XPulsarConsumerThreadContainer.get(i);
            if (xPulsarConsumerThread.getTopic().equals(topic)) {
                xPulsarConsumerThread.stopThread();
                XPulsarConsumerThreadContainer.remove(xPulsarConsumerThread);
            }
        }
        log.info("stop pulsar consumer end . topic={}", topic);
    }

    public ConsumerStats stats(String topic) {
        log.info("stats pulsar consumer start . topic={}", topic);
        for (int i = XPulsarConsumerThreadContainer.size() - 1; i >= 0; i--) {
            XPulsarConsumerThread xPulsarConsumerThread = XPulsarConsumerThreadContainer.get(i);
            if (xPulsarConsumerThread.getTopic().equals(topic)) {
                XPulsarConsumer<String> consumer = xPulsarConsumerThread.getConsumer();
                ConsumerStats stats = consumer.getStats();
                return stats;
            }
        }
        log.info("stats pulsar consumer end . topic={}", topic);
        return null;
    }

    public void stop() {
        log.info("==========CustomerStopEvent===XPulsarConsumerRegister================");
        log.info("shutdownPulsarListener stop");
        pulsarConsumerThreadPool.shutdown();
        for (int i = XPulsarConsumerThreadContainer.size() - 1; i >= 0; i--) {
            XPulsarConsumerThread xPulsarConsumerThread = XPulsarConsumerThreadContainer.get(i);
            xPulsarConsumerThread.stopThread();
            XPulsarConsumerThreadContainer.remove(xPulsarConsumerThread);
        }
        log.info("shutdownPulsarListener end");
    }


}
